

#include "EvaluationRunner.h"

EvaluationRunner::EvaluationRunner(ConfigHandler *configHandler, ParamHandler *parParamHandler, OutputListener *parOutputListener) {
	//context.p = new Producer(parParamHandler);
	//context.engine = new StorexEngine(configHandler, parOutputListener);
	//context.paramHandler = parParamHandler;
	storexEngine = new StorexEngine(configHandler, parOutputListener);
	paramHandler = parParamHandler;
/*
	RulesGenerator rg = RulesGenerator(parParamHandler);
	set<RulePkt *> rulePkts;
	rg.createRulePkts(rulePkts);
	for (set<RulePkt *>::iterator it=rulePkts.begin(); it!=rulePkts.end(); ++it) {
		RulePkt *pkt = *it;
		context.engine->processRulePkt(pkt);
	}
*/
	//context.finish = false;
}

EvaluationRunner::~EvaluationRunner() {
	//delete context.p;
	//delete context.engine;
	delete storexEngine;
}

void * producer(void *parContext) {

/*
	Context *c = (Context *) parContext;
	ParamHandler *paramHandler = c->paramHandler;
	int queueLimit = paramHandler->getQueueSize();
	// Creates and stores publication packets
	queue<PktPtr> pubs;
	int pubsLeft = paramHandler->getPubNum();
	int subsLeft = paramHandler->getSubNum();
	for (int i=0; i<paramHandler->getPubNum()+paramHandler->getSubNum(); i++) {
		if(rand()%2==0){
			if (pubsLeft != 0){
				PktPtr* pkt= new PktPtr();
				*pkt = c->p->createPubPkt();
				pubs.push(*pkt);
				pubsLeft--;
			} else {
				PktPtr* pkt= new PktPtr();
				*pkt = c->p->createBackSubPkt();
				pubs.push(*pkt);
				subsLeft--;
			}
		} else {
			if (subsLeft != 0){
				PktPtr* pkt= new PktPtr();
				*pkt = c->p->createBackSubPkt();
				pubs.push(*pkt);
				subsLeft--;

			} else {
				PktPtr* pkt= new PktPtr();
				*pkt = c->p->createPubPkt();
				pubs.push(*pkt);
				pubsLeft--;
			}
		}
	}
	// Sends packets
	int numDropped = 0;
	Timer t;
	double elapsedTime = 0;
	int extraMessagesToSend = 0;
	int sleepTime = paramHandler->getSleepTime();
	while (! pubs.empty()) {
		pthread_mutex_lock(c->mutex);
		PktPtr pkt = pubs.front();
		pubs.pop();
		if ((int) c->queue.size()>=queueLimit) {
			//delete *pkt;
			numDropped++;
		} else {
			//pkt->setCurrentTime();
			c->queue.push(pkt);
		}
		if (pubs.empty()) {
			c->finish = true;
			c->dropped = numDropped;
		}
		pthread_cond_signal(c->cond);
		pthread_mutex_unlock(c->mutex);
		// Needs to send messages to compensate increased sleeping time
		if (extraMessagesToSend>0) {
			extraMessagesToSend--;
			continue;
		}
		// Sleeps and computes actually sleeping time (extraMessagesToSend is used to compensate rate loss due to thread scheduling)
		t.start();
		usleep(sleepTime);
		t.stop();
		elapsedTime = t.getElapsedTimeInMicroSec();
		if (elapsedTime > sleepTime+500) {
			// TODO: remove comment to generate extraMessages
			//extraMessagesToSend = elapsedTime/sleepTime;
		}
	}
	return NULL;
	*/
}

void * consumer(void *parContext) {
	/*
	Context *c = (Context *) parContext;
	while(true) {
		pthread_mutex_lock(c->mutex);
		if (c->queue.empty()) {
			if (c->finish) {
				pthread_mutex_unlock(c->mutex);
				break;
			}
			pthread_cond_wait(c->cond, c->mutex);
		}
		PktPtr pkt = c->queue.front();
		c->queue.pop();
		pthread_mutex_unlock(c->mutex);
		boost::apply_visitor(PktHandleVisitor(c->engine, c->paramHandler), pkt);
	}
	return NULL;
	*/
}

int EvaluationRunner::startEval(std::queue<PktPtr> inputQueue) {

	cout << endl << "### Starting evaluation ###" << endl << endl;

	while(true) {
		if (inputQueue.empty()) break;
		PktPtr pkt = inputQueue.front();
		inputQueue.pop();
		boost::apply_visitor(PktHandleVisitor(storexEngine, paramHandler), pkt);
	}

	cout << endl << endl << "### Evaluation finished ###" << endl << endl;
	return 0;

/*	context.cond = new pthread_cond_t;
	context.mutex = new pthread_mutex_t;

	pthread_cond_init(context.cond, NULL);
	pthread_mutex_init(context.mutex, NULL);

	pthread_t prod;
	pthread_t cons;

	cout << endl << "### Starting evaluation ###" << endl << endl;

	if (pthread_create(&prod, NULL, producer, (void *) &context)!=0) {
		cout << "Error creating producer" << endl;
		pthread_mutex_destroy(context.mutex);
		pthread_cond_destroy(context.cond);
	}

	if (pthread_create(&cons, NULL, consumer, (void *) &context)!=0) {
		cout << "Error creating consumer" << endl;
		pthread_mutex_lock(context.mutex);
		pthread_mutex_unlock(context.mutex);
		pthread_cond_signal(context.cond);
		pthread_join(prod, NULL);
		pthread_mutex_destroy(context.mutex);
		pthread_cond_destroy(context.cond);;
		return 0;
	}

	pthread_join(prod, NULL);
	pthread_join(cons, NULL);
	pthread_mutex_destroy(context.mutex);
	pthread_cond_destroy(context.cond);
	delete context.cond;
	delete context.mutex;

	cout << endl << endl << "### Evaluation finished ###" << endl << endl;
	return context.dropped;
	*/
}


void PktHandleVisitor::operator()(RulePkt * pkt) const{
	cout << "Received rule: " << test::toString(pkt) << endl;
	engine->processRulePkt(pkt);

}

void PktHandleVisitor::operator()(PubPkt * pkt) const{
	pkt->setCurrentTime();
	cout << "Received message: " << test::toString(pkt) << " with timestamp: " << pkt->getTimeStamp().getTimeVal() << endl;
	engine->processPubPkt(pkt);
}

void PktHandleVisitor::operator()(BackSubPkt * pkt) const{
	pkt->setCurrentTime();
	cout << "Received subscription: " << test::toString(pkt) << " with timestamp: " << pkt->getTimeStamp().getTimeVal() << endl;
	int interface = rand()%paramHandler->getNumInterfaces();
	Inbox* inbox = new Inbox(interface, pkt);
	engine->processSubPkt(inbox);
}

void PktHandleVisitor::operator()(UnSubPkt * pUnsub) const{

}

void PktHandleVisitor::operator()(SubPkt * pkt) const{

}

void PktHandleVisitor::operator()(PingPkt * pkt) const{

}
